package com.pig4cloud.pigx.contract.api.v1.websocket.subscribe;

import com.alibaba.fastjson.JSONObject;
import com.pig4cloud.pigx.common.core.util.R;
import com.pig4cloud.pigx.common.customize.api.enums.AbnormalEnum;
import com.pig4cloud.pigx.common.customize.api.exception.OpenApiException;
import com.pig4cloud.pigx.common.customize.i18n.util.MessageUtils;
import com.pig4cloud.pigx.contract.api.v1.utils.OpenApiContractUtilsV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.dto.OpenApiSymbolDTOV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.dto.OpenApiWebSocketMessageDTOV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.message.OpenApiSendMessageV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.vo.OpenApi24hTickerVOV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.vo.OpenApiAbstractTopicVOV1;
import com.pig4cloud.pigx.contract.mongo.entity.ContractSymbolMongo;
import com.pig4cloud.pigx.contract.utils.ContractUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 */
@Slf4j
@Component
public class OpenApiSymbol24hTickerV1 extends RequestDataAbstractTopicV1 {

    private final OpenApiContractUtilsV1 apiContractUtils;

    private final ContractUtils contractUtils;

    public static final String API_NAME = "symbol_24h_ticker";

    public OpenApiSymbol24hTickerV1(OpenApiContractUtilsV1 apiContractUtils, ContractUtils contractUtils) {
        this.apiContractUtils = apiContractUtils;
        this.contractUtils = contractUtils;
        init();
    }

    private void init() {
        wsPush();
    }

    public void nowSub(WebSocketSession session, OpenApiWebSocketMessageDTOV1 params) {
        // 接受新消息时,立刻处理 给予响应
        OpenApiAbstractTopicVOV1 sub = sub(session, params);
        Map<WebSocketSession, OpenApiAbstractTopicVOV1> map = new ConcurrentHashMap<>(1);
        map.put(session, sub);
        sendMessage(map);
    }

    private void wsPush() {
        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
        service.scheduleAtFixedRate(() -> {
            sendMessage(subList);
        }, 0, 1000, TimeUnit.MILLISECONDS);
    }

    private void sendMessage(Map<WebSocketSession, OpenApiAbstractTopicVOV1> list) {
        Map<String, List<WebSocketSession>> map = new HashMap<>();
        // 记录请求参数不合法的session,发送一次报错信息后,移除session队列
        List<WebSocketSession> requestErrorList = new ArrayList<>();
        // 根据参数对session分组
        for (Map.Entry<WebSocketSession, OpenApiAbstractTopicVOV1> entry : list.entrySet()) {
            if (!entry.getKey().isOpen()) {
                subList.remove(entry.getKey());
                log.info("ws 移除已关闭的session");
                continue;
            }
            try {
                // 开始参数校验
                JSONObject param = entry.getValue().getData().getRequestParam();
                OpenApiSymbolDTOV1 dto = JSONObject.parseObject(param.toJSONString(), OpenApiSymbolDTOV1.class);
                if (dto == null || CollectionUtils.isEmpty(dto.getSymbolList())) {
                    // 必填参数为空发送异常消息
                    entry.getKey().sendMessage(OpenApiSendMessageV1.getTextMessage(R.failed(
                            AbnormalEnum.MSG.MANDATORY_PARAM_EMPTY_OR_MALFORMED.getCode(),
                            MessageUtils.get(AbnormalEnum.MSG.MANDATORY_PARAM_EMPTY_OR_MALFORMED.getMessage()))
                    ));
                    requestErrorList.add(entry.getKey());
                    continue;
                }

                // 以币对名称为key对用户分组
                dto.getSymbolList().forEach(symbolName -> map.computeIfAbsent(symbolName, key -> new ArrayList<>())
                        .add(entry.getKey()));

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        // 移除请求参数异常的session
        for (WebSocketSession session : requestErrorList) {
            unSub(session);
        }
        try {
            // 记录请求异常的session,发送一次报错信息后,移除session队列
            List<WebSocketSession> symbolErrorList = new ArrayList<>();
            for (Map.Entry<String, List<WebSocketSession>> entry : map.entrySet()) {
                String key = entry.getKey();
                List<WebSocketSession> value = entry.getValue();
                // 币对校验
                ContractSymbolMongo mongo;
                try {
                    // 币对校验
                    mongo = apiContractUtils.getSymbolMongo(key);
                    // 类型转换 精度设置
                    OpenApi24hTickerVOV1 vo = new OpenApi24hTickerVOV1();
                    BeanUtils.copyProperties(mongo, vo);
                    contractUtils.setScaleBySymbol(vo, mongo.getName());
                    vo.setSymbolName(key);
                    for (WebSocketSession session : value) {
                        try {
                            // 发送币对最新价
                            session.sendMessage(OpenApiSendMessageV1.getTextMessage(
                                    R.ok(vo)
                            ));
                        } catch (IOException exception) {
                            throw new RuntimeException(exception);
                        }
                    }
                } catch (OpenApiException e) {
                    // 返回异常消息
                    for (WebSocketSession session : value) {
                        try {
                            // 将捕获的异常消息发送
                            session.sendMessage(OpenApiSendMessageV1.getTextMessage(
                                    R.failed(e.getCode(), e.getMessage())
                            ));
                            symbolErrorList.add(session);
                        } catch (IOException exception) {
                            throw new RuntimeException(exception);
                        }
                    }
                }
            }
            // 移除请求无效的session
            for (WebSocketSession session : symbolErrorList) {
                unSub(session);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("webSocket---symbol_last_price---最新币对24hTicker Error");
        }
    }

}
